package com.lagou.demo6;
/*
依赖说明：
flink-table-api-java-bridge_2.1：桥接器，主要负责 table API 和 DataStream/DataSet API
的连接支持，按照语言分 java 和 scala。
flink-table-planner-blink_2.12：计划器，是 table API 最主要的部分，提供了运行时环境和生
成程序执行计划的 planner；
如果是生产环境，lib 目录下默认已 经有了 planner，就只需要有 bridge 就可以了
flink-table：flinktable的基础依赖
 */

/*
flinktable案例代码编写
作业提交流程图自己亲自重绘
1、Flink执行环境env
2、用env，做出Table环境tEnv
3、获取流式数据源
4、将流式数据源做成Table
5、对Table中的数据做查询
6、将Table转成数据流
 */


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;

import static org.apache.flink.table.api.Expressions.$;


/*
Flink依赖的scope默认为provided, 因此IDEA不会加载这些依赖
解决方法：修改Run/Debug Configurations,勾选Include dependencies with "provided" scope
 */
public class TableApiDemo {
    public static void main(String[] args) throws Exception {
        //1、Flink执行环境env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2、用env，做出Table环境tEnv
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //3、获取流式数据源
        DataStreamSource<Tuple2<String, Integer>> data = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                ctx.collect(new Tuple2<>("name", 10));
                Thread.sleep(1000);
            }

            @Override
            public void cancel() {

            }
        });


        //4、将流式数据源做成Table
        Table table = tEnv.fromDataStream(data, $("name"), $("age"));

        //5、对Table中的数据做查询
        Table name = table.select($("name"));

        //6、将处理结果输出到控制台
        DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(name, Row.class);

        result.print();
        env.execute();
    }


}
